package com.settlement.system.service.impl;

import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.settlement.system.common.base.IBaseEnum;
import com.settlement.system.common.constant.KafkaConstants;
import com.settlement.system.common.constant.SystemConstants;
import com.settlement.system.common.enums.DeleteEnum;
import com.settlement.system.common.enums.NoMatchRecordStatusEnum;
import com.settlement.system.common.enums.SourceBillToQueueEnum;
import com.settlement.system.common.exception.BusinessException;
import com.settlement.system.common.util.StringUtils;
import com.settlement.system.handler.disruptor.MessageVo;
import com.settlement.system.mapper.NoMatchRecordMapper;
import com.settlement.system.mapper.SysWorkingQueueMapper;
import com.settlement.system.mapper.TmFundFlowMapper;
import com.settlement.system.model.entity.*;
import com.settlement.system.model.form.NoMatchRecordForm;
import com.settlement.system.model.query.NoMatchRecordQuery;
import com.settlement.system.model.query.SysWorkingQueueQuery;
import com.settlement.system.model.vo.NoMatchRecordImportVo;
import com.settlement.system.model.vo.TmFundFlowImportVo;
import com.settlement.system.service.*;
import lombok.RequiredArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

/**
 * <p>
 * 转换不到账单记录表 服务实现类
 * </p>
 *
 * @author huangrensen
 * @since 2024-08-09
 */
@Service
@RequiredArgsConstructor
public class NoMatchRecordServiceImpl extends ServiceImpl<NoMatchRecordMapper, NoMatchRecord> implements NoMatchRecordService {

    private final static Logger logger = LoggerFactory.getLogger(NoMatchRecordServiceImpl.class);
    private final SysStoreService storeService;
    private final KafkaTemplate<String, String> kafkaTemplate;

    private final SysWorkingQueueMapper workingQueueMapper;

    private final SysWorkingQueueService workingQueueService;

    private final RedisTemplate redisTemplate;

    private final TmFundFlowMapper tmFundFlowMapper;

    private final KafkaSendErrorRecordService kafkaSendErrorRecordService;
    private final SnowflakeIdWorker snowflakeIdWorker = SnowflakeIdWorker.getInstance();

    @Override
    public Page<NoMatchRecord> getPage(NoMatchRecordQuery queryParams) {
        // 查询参数
        int pageNum = queryParams.getPageNum();
        int pageSize = queryParams.getPageSize();
        // 查询数据
        if (Objects.nonNull(queryParams.getEndDate())) {
            queryParams.setEndDate(queryParams.getEndDate().plusDays(1));
        }
        Page<NoMatchRecord> page = new Page<>(pageNum, pageSize);
        Page<NoMatchRecord> mapperPage = this.baseMapper.getPage(page, queryParams);
        if (mapperPage.getRecords().size() > 0) {
            SysStore store;
            String sourceData;
            for (NoMatchRecord record : mapperPage.getRecords()) {
                store = storeService.getById(record.getStoreId());
                if (Objects.nonNull(store)) {
                    record.setStoreName(store.getCommercePlatformStoreName());
                }
                record.setStatusName(IBaseEnum.getLabelByValue(record.getStatus(), NoMatchRecordStatusEnum.class));
                // 查询原始数据
                sourceData = this.getDataInfo(record);
                if(StringUtils.hasText(sourceData)){
                    record.setDataMap(sourceData);
                }
            }
        }
        return mapperPage;
    }

    @Override
    public boolean saveNoMatchRecord(NoMatchRecordForm noMatchRecordForm) {
        NoMatchRecord noMatchRecord = new NoMatchRecord();
        BeanUtils.copyProperties(noMatchRecordForm, noMatchRecord);
        noMatchRecord.setCreateTime(null);
        return this.saveOrUpdate(noMatchRecord);
    }

    @Override
    public NoMatchRecordForm getNoMatchRecordFormData(Long id) {
        NoMatchRecord noMatchRecord = this.getById(id);
        NoMatchRecordForm noMatchRecordForm = new NoMatchRecordForm();
        BeanUtils.copyProperties(noMatchRecord, noMatchRecordForm);
        return noMatchRecordForm;
    }

    @Override
    public boolean deleteNoMatchRecord(String ids) {
        Assert.isTrue(StringUtils.hasText(ids), "删除的数据为空");
        List<Long> idArray = Arrays.stream(ids.split(",")).map(Long::parseLong).collect(Collectors.toList());
        return this.removeByIds(idArray);
    }

    @Override
    public void importData(List<NoMatchRecordImportVo> voList) {
        List<NoMatchRecord> noMatchRecordList = new ArrayList<>();
        for (NoMatchRecordImportVo importVo : voList) {
            NoMatchRecord noMatchRecord = new NoMatchRecord();
            BeanUtils.copyProperties(importVo, noMatchRecord);
            noMatchRecordList.add(noMatchRecord);
        }
        this.saveOrUpdateBatch(noMatchRecordList);
    }

    @Override
    public List<NoMatchRecordImportVo> findByIds(String idsStr) {
        if (!StringUtils.hasText(idsStr)) {
            return new ArrayList<>();
        }
        List<Long> ids = Arrays.stream(idsStr.split(",")).map(Long::parseLong).collect(Collectors.toList());
        List<NoMatchRecord> noMatchRecordList = this.getBaseMapper().selectBatchIds(ids);
        List<NoMatchRecordImportVo> voList = new ArrayList<>();
        for (NoMatchRecord noMatchRecord : noMatchRecordList) {
            NoMatchRecordImportVo vo = new NoMatchRecordImportVo();
            BeanUtils.copyProperties(noMatchRecord, vo);
            voList.add(vo);
        }
        return voList;
    }

    @Override
    public void reBill(String idsStr) {
        if (!StringUtils.hasText(idsStr)) {
            return;
        }
        List<Long> ids = Arrays.stream(idsStr.split(",")).map(Long::parseLong).collect(Collectors.toList());
        List<NoMatchRecord> noMatchRecordList = this.getBaseMapper().selectBatchIds(ids);
        MessageVo vo;
        for (NoMatchRecord noMatchRecord : noMatchRecordList) {
            // 检查是否存在原始数据
            if (!hasSourceData(noMatchRecord)) {
                // 原始数据已经删除
                noMatchRecord.setDeleted(1);
                continue;
            }
            // 发送通知
            vo = new MessageVo();
            vo.setDataId(noMatchRecord.getDataId());
            vo.setStoreId(noMatchRecord.getStoreId());
            vo.setBillMonth(noMatchRecord.getBillMonth());
            // 根据字典中的备注查找对应的数据库表字段名
            vo.setTableNamePrefix(noMatchRecord.getTableNamePreFix());
            // 从字段管理里找到对应的字段类型编码，获取对应的billType
            vo.setDictCode(noMatchRecord.getDictCode());
            // 转换失败记录ID
            vo.setNoMatchRecordId(noMatchRecord.getId());
            // 重新发送消息 推送到转换标准账单的消息队列
            vo.setUuid(snowflakeIdWorker.nextId());
            CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_BILL_QUEUE, JSONObject.toJSONString(vo));
            MessageVo finalVo = vo;
            completableFuture.whenComplete((result, ex) -> {
                if(Objects.nonNull(ex)){
                    ex.printStackTrace();
                    KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                    errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                    errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_BILL_QUEUE);
                    errorRecord.setRemark(ex.getMessage());
                    errorRecord.setUuid(finalVo.getUuid());
                    kafkaSendErrorRecordService.checkAndSave(errorRecord);
                }
            });
            // 转换中
            noMatchRecord.setStatus(NoMatchRecordStatusEnum.RUNNING.getValue());
            noMatchRecord.setUpdateTime(LocalDateTime.now());
        }
        this.updateBatchById(noMatchRecordList);
    }

    @Override
    public void ignore(String idsStr) {
        if (!StringUtils.hasText(idsStr)) {
            return;
        }
        List<Long> ids = Arrays.stream(idsStr.split(",")).map(Long::parseLong).collect(Collectors.toList());
        List<NoMatchRecord> noMatchRecordList = this.getBaseMapper().selectBatchIds(ids);
        noMatchRecordList.forEach(item -> item.setDeleted(DeleteEnum.YES.getValue()));
        this.updateBatchById(noMatchRecordList);
    }

    @Override
    public void reBillAll(NoMatchRecordQuery queryParams) {
        Object lockObj = redisTemplate.opsForValue().get(SystemConstants.WORKING_QUEUE_LOCK);
        Assert.isNull(lockObj, "有任务正在处理，请稍等几分钟再试...");
        // 锁住队列
        redisTemplate.opsForValue().set(SystemConstants.WORKING_QUEUE_LOCK, SystemConstants.WORKING_QUEUE_LOCK);
        // 开始工作
        boolean runFlag;
        try {
            if (StringUtils.hasText(workingQueueMapper.existTable())) {
                workingQueueMapper.dropTable();
            }
            // 创建表
            workingQueueMapper.createTable();
            queryParams.setPageSize(300);
            queryParams.setPageNum(1);
            // 只查失败的记录
            queryParams.setStatus(NoMatchRecordStatusEnum.ERROR.getValue());
            logger.info("正在查询账单转换失败的记录...");
            Page<NoMatchRecord> page = this.getPage(queryParams);
            List<SysWorkingQueue> workingQueues;
            SysWorkingQueue workingQueue;
            while (page.getRecords().size() > 0) {
                // 正在把数据转移到另外一张表中
                workingQueues = new ArrayList<>();
                for (NoMatchRecord record : page.getRecords()) {
                    workingQueue = new SysWorkingQueue();
                    workingQueue.setDataId(record.getId().toString());
                    workingQueues.add(workingQueue);
                }
                // 保存数据
                workingQueueService.saveBatch(workingQueues);
                // 查询下一页
                queryParams.setPageNum(queryParams.getPageNum() + 1);
                logger.info("正在查询{}页数据，共{}页...", queryParams.getPageNum(), page.getPages());
                page = this.getPage(queryParams);
            }
            logger.info("查询账单转换失败的记录完成...");
            // 然后直接查询工作表的数据即可
            SysWorkingQueueQuery workingQueueQuery = new SysWorkingQueueQuery();
            workingQueueQuery.setPageNum(1);
            workingQueueQuery.setPageSize(300);
            Page<SysWorkingQueue> workingQueuePage = workingQueueService.getPage(workingQueueQuery);
            List<String> dataIdList;
            List<NoMatchRecord> noMatchRecordList;
            while (workingQueuePage.getRecords().size() > 0) {
                // 再按照ID列表去查
                dataIdList = workingQueuePage.getRecords().stream().map(SysWorkingQueue::getDataId).toList();
                noMatchRecordList = this.getBaseMapper().selectBatchIds(dataIdList);
                logger.info("正在推送第{}页的数据，共{}页...", workingQueueQuery.getPageNum(), workingQueuePage.getPages());
                this.pushToQueue(noMatchRecordList);
                logger.info("推送第{}页的数据完成，共{}页...", workingQueueQuery.getPageNum(), workingQueuePage.getPages());
                // 查询下一页
                workingQueueQuery.setPageNum(workingQueueQuery.getPageNum() + 1);
                workingQueuePage = workingQueueService.getPage(workingQueueQuery);
            }
            runFlag = true;
        } catch (Exception e) {
            e.printStackTrace();
            runFlag = false;
        } finally {
            // 删除锁住标志
            redisTemplate.delete(SystemConstants.WORKING_QUEUE_LOCK);
        }
        // 检查是否出现错误
        if (!runFlag) {
            throw new BusinessException("重新转换原始账单失败,请联系运维人员");
        }
    }

    private void pushToQueue(List<NoMatchRecord> noMatchRecordList){
        List<MessageVo> voList = new ArrayList<>();
        MessageVo vo;
        for (NoMatchRecord noMatchRecord : noMatchRecordList) {
            // 发送通知
            vo = new MessageVo();
            vo.setDataId(noMatchRecord.getDataId());
            vo.setStoreId(noMatchRecord.getStoreId());
            vo.setBillMonth(noMatchRecord.getBillMonth());
            // 根据字典中的备注查找对应的数据库表字段名
            vo.setTableNamePrefix(noMatchRecord.getTableNamePreFix());
            // 从字段管理里找到对应的字段类型编码，获取对应的billType
            vo.setDictCode(noMatchRecord.getDictCode());
            // 转换失败记录ID
            vo.setNoMatchRecordId(noMatchRecord.getId());
            // 转换中
            noMatchRecord.setStatus(NoMatchRecordStatusEnum.RUNNING.getValue());
            noMatchRecord.setUpdateTime(LocalDateTime.now());
            // 添加到列表
            voList.add(vo);
        }
        // 更新状态
        this.updateBatchById(noMatchRecordList);
        for (MessageVo messageVo : voList) {
            // 重新发送消息 推送到转换标准账单的消息队列
            messageVo.setUuid(snowflakeIdWorker.nextId());
            CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_BILL_QUEUE, JSONObject.toJSONString(messageVo));
            completableFuture.whenComplete((result, throwable) -> {
                if (Objects.nonNull(throwable)) {
                    throwable.printStackTrace();
                    logger.error("推送消息失败，消息ID:{},失败原因:{}", messageVo.getUuid(), throwable.getMessage());
                    KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                    errorRecord.setDataText(JSONObject.toJSONString(messageVo));
                    errorRecord.setRemark(throwable.getMessage());
                    errorRecord.setUuid(messageVo.getUuid());
                    errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_BILL_QUEUE);
                    kafkaSendErrorRecordService.checkAndSave(errorRecord);
                }
            });
        }
    }

    public boolean hasSourceData(NoMatchRecord noMatchRecord) {
        String tableName = "%s%s".formatted(noMatchRecord.getTableNamePreFix(), noMatchRecord.getBillMonth());
        String sql = "select id from `%s` where id = '%s'".formatted(tableName, noMatchRecord.getDataId());
        List<Map<String, Object>> mapList = this.baseMapper.getDataBySql(sql);
        return mapList.size() > 0;
    }

    @Override
    public String getDataInfo(Long id) {
        NoMatchRecord noMatchRecord = this.getById(id);
        String tableName = "%s%s".formatted(noMatchRecord.getTableNamePreFix(), noMatchRecord.getBillMonth());
        String sql = "select * from `%s` where id = '%s'".formatted(tableName, noMatchRecord.getDataId());
        List<Map<String, Object>> mapList = this.baseMapper.getDataBySql(sql);
        if (mapList.size() > 0) {
            return JSONObject.toJSONString(mapList.get(0));
        }
        return null;
    }

    private String getDataInfo(NoMatchRecord noMatchRecord) {
        String tableName = "%s%s".formatted(noMatchRecord.getTableNamePreFix(), noMatchRecord.getBillMonth());
        String sql = "select * from `%s` where id = '%s'".formatted(tableName, noMatchRecord.getDataId());
        List<Map<String, Object>> mapList = this.baseMapper.getDataBySql(sql);
        if (mapList.size() > 0) {
            return JSONObject.toJSONString(mapList.get(0));
        }
        return null;
    }

    @Override
    public void fallback(NoMatchRecordQuery queryParams) {
        String queueName = IBaseEnum.getLabelByValue(queryParams.getBillType(), SourceBillToQueueEnum.class);
        Assert.notNull(queueName, "此账单类型【%s】没有配置到队列绑定关系".formatted(queryParams.getBillType()));
        queryParams.setPageNum(1);
        queryParams.setPageSize(300);
        queryParams.setStatus(NoMatchRecordStatusEnum.ERROR.getValue());
        Page<NoMatchRecord> page = this.getPage(queryParams);
        List<NoMatchRecord> noMatchRecordList;
        TmFundFlow tmFundFlow;
        while (page.getRecords().size() > 0) {
            noMatchRecordList = page.getRecords();
            for (NoMatchRecord noMatchRecord : noMatchRecordList) {
                // 转换状态
                tmFundFlow = tmFundFlowMapper.getDataById(noMatchRecord.getTableNamePreFix() + noMatchRecord.getBillMonth(), noMatchRecord.getDataId());
                if (Objects.nonNull(tmFundFlow)) {
                    // 转换单号
                    tmFundFlow.formatSellerOrderNo();
                    noMatchRecord.setSellerOrderNo(tmFundFlow.getSellerOrderNo());
                    tmFundFlow.setTableName(noMatchRecord.getTableNamePreFix() + noMatchRecord.getBillMonth());
                    tmFundFlowMapper.update(tmFundFlow);
                } else {
                    // 原始单据不存在，标记删除
                    noMatchRecord.setDeleted(DeleteEnum.YES.getValue());
                }
            }
            // 更新
            this.updateBatchById(noMatchRecordList);
            // 查询下一页
            queryParams.setPageNum(queryParams.getPageNum() + 1);
            page = this.getPage(queryParams);
        }
        this.reBillAll(queryParams);
    }
}
